package trident.partition;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class PartitionDemo {
    public static void main(String[] args) throws Exception {
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("name", "hive", "storm"), 4,
                new Values("A", 80, 60),
                new Values("V", 80, 40),
                new Values("C", 50, 80),
                new Values("B", 70, 60),
                new Values("E", 80, 90),
                new Values("W", 30, 50),
                new Values("M", 90, 60),
                new Values("K", 60, 90),
                new Values("S", 70, 50)
        );

        spout.setCycle(false); // 循环发射数据

        // 构建器 Builder
        TridentTopology tridentTopology = new TridentTopology();
        // 绑定数据源
        tridentTopology.newStream("spout", spout)
                .each(new Fields("hive", "storm"), new ScoreFilter())
                .each(new Fields("hive", "storm"), new AverageFunction(), new Fields("average"))
                .project(new Fields("name", "average")) // 过滤字段：仅保留这两个字段
                .each(new Fields("name", "average"), new PrintFunction(), new Fields())
        ;

        // 提交topology
        Config config = new Config();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("partition-demo", config, tridentTopology.build());
    }
}
